package cn.spark.study.core;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.List;

/**
 * 取top3
 *
 * @author jun.zhang6
 * @date 2020/11/6
 */
public class Top3 {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf()
                .setAppName("Top3")
                .setMaster("local");

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD<String> numbersRdd = sc.textFile("C:\\Users\\jun.zhang6\\Desktop\\top.txt");

        JavaPairRDD<Integer, String> pairsRdd = numbersRdd.mapToPair(new PairFunction<String, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(String v) throws Exception {
                return new Tuple2<Integer, String>(Integer.valueOf(v), v);
            }
        });

        JavaPairRDD<Integer, String> sortedRdd = pairsRdd.sortByKey(false);

        JavaRDD<String> mapedRdd = sortedRdd.map(new Function<Tuple2<Integer, String>, String>() {
            @Override
            public String call(Tuple2<Integer, String> tuple) throws Exception {
                return tuple._2();
            }
        });

        List<String> top3List = mapedRdd.take(3);

        for (String num : top3List) {
            System.out.println(num);
        }

        sc.close();

    }
}
